package com.wangwg.sparkTest.hdfs

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import org.geotools.data.Transaction
import org.geotools.feature.simple.SimpleFeatureTypeBuilder
import org.locationtech.geomesa.fs.storage.common.RichSimpleFeatureType
import org.locationtech.geomesa.fs.storage.common.interop.ConfigurationUtils
import org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator
import org.locationtech.jts.geom.Polygon

import scala.collection.JavaConversions._

object GeomesaHdfsCreate {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("GeomesaHdfsCreate");
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    sparkConf.set("spark.kryo.registrator", classOf[GeoMesaSparkKryoRegistrator].getName)
    val sparkContext = SparkContext.getOrCreate(sparkConf);

    val hdfsUrl = "hdfs://lhyg-wangwengang:9000"
    val hdfsConf: Configuration = new Configuration();
    hdfsConf.set("fs.defaultFS", hdfsUrl)
    val hdfs = FileSystem.get(hdfsConf)
    val shpName = "landuse"
    hdfs.mkdirs(new Path("/input/".concat(shpName)))
    
    val path = new Path("/input/".concat(shpName))

    var outputParams: Map[String, String] = Map()
    outputParams += ("fs.path" -> path.toUri.toString)
    val fileName = shpName.concat("_").concat(TaskContext.getPartitionId().toString)
    val dataStore = org.geotools.data.DataStoreFinder.getDataStore(outputParams)

    val simpleFeatureTypeBuilder = new SimpleFeatureTypeBuilder
    simpleFeatureTypeBuilder.setName(fileName)
    simpleFeatureTypeBuilder.add("the_geom", classOf[Polygon])
    simpleFeatureTypeBuilder.add("dkbh", classOf[String])
    simpleFeatureTypeBuilder.add("dkmc", classOf[String])

    val simpleFeatureType = simpleFeatureTypeBuilder.buildFeatureType()
    //simpleFeatureType.setScheme(shpName,Map("dkmc"->""))
    simpleFeatureType.setScheme("daily", Map("dtg-attribute" -> "dtg"))
    ConfigurationUtils.setEncoding(simpleFeatureType, "UTF-8")
    //ConfigurationUtils.setScheme(simpleFeatureType, ,Collections.emptyMap)
    dataStore.createSchema(simpleFeatureType)

    val featureWriter = dataStore.getFeatureWriter(shpName, Transaction.AUTO_COMMIT)
    featureWriter.close()
  }
}
